import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{SaveMode, SparkSession}
import org.apache.spark.{SparkConf, SparkContext}
import org.json4s.JsonDSL._
import org.json4s.jackson.JsonMethods._

import java.io.PrintWriter

/**
 * 根据血糖值判定是否为糖尿病
 * 正常：3.9-6.1
 * 糖尿病前期：6.1-7.0
 * 糖尿病：大于7.0
 */
case class MeiResult(binglitype:String,mei:String,number:Double)
object meiAnalyze {
  def main(args: Array[String]): Unit = {
    //1.创建Spark环境配置 val sparkConf = new SparkConf().setAppName("AgeStageAnalyze").setMaster("local")
    val sparkConf = new SparkConf().setAppName("meiAnalyze").setMaster("local")
    //2.创建SparkContext上下文环境
    val sc: SparkContext = new SparkContext(sparkConf)
    val spark: SparkSession = SparkSession.builder().config(sparkConf).getOrCreate()

    import spark.implicits._
    //3.定义数据源文件
    //  保存数据分析结果的文件
    val inputFile = "src\\output\\cleaned\\tnb_cleaned.csv\\part-00000-7da365d4-d48c-400d-b8b4-29c7abc55d37-c000.csv"
    val outputFile = "src\\output\\mei1.json"
    //hadoop集群调试
    //    val inputFile = args(0)
    //    val outputFile =args(1)
    //4. 读取数据源文件
    //   4.1 如果源文件第一行表示的是不同列的属性值，则在数据分析时应过滤掉第一行内容;如果没有属性行，则不需要过滤
    val userinfodata: RDD[String] = sc.textFile(inputFile)

    var cleanUserinfoData: RDD[String] = userinfodata.filter(x => {
      val userinfoSplitData: Array[String] = x.split(",")
      x.startsWith("id,性别") == false
    })

    /*userinfodata.filter( x=> x.startsWith("userId")==false)
      .filter(x=> {val line =x.split(","); line.length>=13})*/


    //   4.2 只获取age特征值
    val ageData = cleanUserinfoData.map(x => {
      val line = x.split(",")
      (line(3).toDouble, line(4).toDouble, line(5).toDouble, line(6).toDouble,
        line(7).toDouble, line(8).toDouble, line(9).toDouble, line(10).toDouble, line(11).toDouble)
    })
    //    ageData.take(6).foreach(println)

    //5. 数据分析
    //5.1   age:(12,34,26,21,....)
    //5.2   按年龄段分析统计
    //            reduceByKey
    val zhengchang = ageData.filter(x => x._9 >= 3.9 & x._9 <= 6.1)
    val qianqi = ageData.filter(x => x._9 > 6.1 & x._9 <= 7.0)
    val zhongdu = ageData.filter(x => x._9 > 7.0)
    //    val zhengchang_total: List[Int] = zhengchang.map(x => ("正常", 1)).reduceByKey((x, y) => x + y).map(x => x._2).collect().toList
    ////    val zhengchangnum=
    val zhengchangnum = zhengchang.count().toDouble
    println("zhengccccccccccccccccccc" + zhengchangnum)
    val qianqinum = qianqi.count()
    val zhongdunum = zhongdu.count()
    val zhengchangresult: RDD[(String, Double)] =
      zhengchang.map(x => ("天门冬氨酸氨基转换酶", x._1)).reduceByKey((x, y) => x + y).map(x => ("天门冬氨酸氨基转换酶", (x._2 / zhengchangnum).formatted("%.2f").toDouble))
        .union(zhengchang.map(x => ("丙氨酸氨基转换酶", x._2)).reduceByKey((x, y) => x + y).map(x => ("丙氨酸氨基转换酶", (x._2 / zhengchangnum).formatted("%.2f").toDouble)))
        .union(zhengchang.map(x => ("碱性磷酸酶", x._3)).reduceByKey((x, y) => x + y).map(x => ("碱性磷酸酶", (x._2 / zhengchangnum).formatted("%.2f").toDouble)))
        .union(zhengchang.map(x => ("谷氨酰基转换酶", x._4)).reduceByKey((x, y) => x + y).map(x => ("谷氨酰基转换酶", (x._2 / zhengchangnum).formatted("%.2f").toDouble)))
        .union(zhengchang.map(x => ("甘油三酯", x._5)).reduceByKey((x, y) => x + y).map(x => ("甘油三酯", (x._2 / zhengchangnum).formatted("%.2f").toDouble)))
        .union(zhengchang.map(x => ("总胆固醇", x._6)).reduceByKey((x, y) => x + y).map(x => ("总胆固醇", (x._2 / zhengchangnum).formatted("%.2f").toDouble)))
        .union(zhengchang.map(x => ("高密度脂蛋白胆固醇", x._7)).reduceByKey((x, y) => x + y).map(x => ("高密度脂蛋白胆固醇", (x._2 / zhengchangnum).formatted("%.2f").toDouble)))
        .union(zhengchang.map(x => ("低密度脂蛋白胆固醇", x._8)).reduceByKey((x, y) => x + y).map(x => ("低密度脂蛋白胆固醇", (x._2 / zhengchangnum).formatted("%.2f").toDouble)))
        .union(zhengchang.map(x => ("血糖", x._9)).reduceByKey((x, y) => x + y).map(x => ("血糖", (x._2 / zhengchangnum).formatted("%.2f").toDouble)))
    val qianqiresult: RDD[(String, Double)] =
      qianqi.map(x => ("天门冬氨酸氨基转换酶", x._1)).reduceByKey((x, y) => x + y).map(x => ("天门冬氨酸氨基转换酶", (x._2 / qianqinum).formatted("%.2f").toDouble))
        .union(qianqi.map(x => ("丙氨酸氨基转换酶", x._2)).reduceByKey((x, y) => x + y).map(x => ("丙氨酸氨基转换酶", (x._2 / qianqinum).formatted("%.2f").toDouble)))
        .union(qianqi.map(x => ("碱性磷酸酶", x._3)).reduceByKey((x, y) => x + y).map(x => ("碱性磷酸酶", (x._2 / qianqinum).formatted("%.2f").toDouble)))
        .union(qianqi.map(x => ("谷氨酰基转换酶", x._4)).reduceByKey((x, y) => x + y).map(x => ("谷氨酰基转换酶", (x._2 / qianqinum).formatted("%.2f").toDouble)))
        .union(qianqi.map(x => ("甘油三酯", x._5)).reduceByKey((x, y) => x + y).map(x => ("甘油三酯", (x._2 / qianqinum).formatted("%.2f").toDouble)))
        .union(qianqi.map(x => ("总胆固醇", x._6)).reduceByKey((x, y) => x + y).map(x => ("总胆固醇", (x._2 / qianqinum).formatted("%.2f").toDouble)))
        .union(qianqi.map(x => ("高密度脂蛋白胆固醇", x._7)).reduceByKey((x, y) => x + y).map(x => ("高密度脂蛋白胆固醇", (x._2 / qianqinum).formatted("%.2f").toDouble)))
        .union(qianqi.map(x => ("低密度脂蛋白胆固醇", x._8)).reduceByKey((x, y) => x + y).map(x => ("低密度脂蛋白胆固醇", (x._2 / qianqinum).formatted("%.2f").toDouble)))
        .union(qianqi.map(x => ("血糖", x._9)).reduceByKey((x, y) => x + y).map(x => ("血糖", (x._2 / qianqinum).formatted("%.2f").toDouble)))
    val zhongduresult: RDD[(String, Double)] =
      zhongdu.map(x => ("天门冬氨酸氨基转换酶", x._1)).reduceByKey((x, y) => x + y).map(x => ("天门冬氨酸氨基转换酶", (x._2 / zhongdunum).formatted("%.2f").toDouble))
        .union(zhongdu.map(x => ("丙氨酸氨基转换酶", x._2)).reduceByKey((x, y) => x + y).map(x => ("丙氨酸氨基转换酶", (x._2 / zhongdunum).formatted("%.2f").toDouble)))
        .union(zhongdu.map(x => ("碱性磷酸酶", x._3)).reduceByKey((x, y) => x + y).map(x => ("碱性磷酸酶", (x._2 / zhongdunum).formatted("%.2f").toDouble)))
        .union(zhongdu.map(x => ("谷氨酰基转换酶", x._4)).reduceByKey((x, y) => x + y).map(x => ("谷氨酰基转换酶", (x._2 / zhongdunum).formatted("%.2f").toDouble)))
        .union(zhongdu.map(x => ("甘油三酯", x._5)).reduceByKey((x, y) => x + y).map(x => ("甘油三酯", (x._2 / zhongdunum).formatted("%.2f").toDouble)))
        .union(zhongdu.map(x => ("总胆固醇", x._6)).reduceByKey((x, y) => x + y).map(x => ("总胆固醇", (x._2 / zhongdunum).formatted("%.2f").toDouble)))
        .union(zhongdu.map(x => ("高密度脂蛋白胆固醇", x._7)).reduceByKey((x, y) => x + y).map(x => ("高密度脂蛋白胆固醇", (x._2 / zhongdunum).formatted("%.2f").toDouble)))
        .union(zhongdu.map(x => ("低密度脂蛋白胆固醇", x._8)).reduceByKey((x, y) => x + y).map(x => ("低密度脂蛋白胆固醇", (x._2 / zhongdunum).formatted("%.2f").toDouble)))
        .union(zhongdu.map(x => ("血糖", x._9)).reduceByKey((x, y) => x + y).map(x => ("血糖", (x._2 / zhongdunum).formatted("%.2f").toDouble)))
    val zhengchangDF = zhengchangresult.map(x => {
      MeiResult("正常", x._1, x._2)
    }).toDF()
    val qianqiDF = qianqiresult.map(x => {
      MeiResult("糖尿病前期患者", x._1, x._2)
    }).toDF()
    val zhongduDF = zhongduresult.map(x => {
      MeiResult("糖尿病重度患者", x._1, x._2)
    }).toDF()
    zhengchangDF.write
      .format("jdbc")
      .option("url", "jdbc:mysql://localhost/health_monitoring")
      .option("user", "root")
      .option("password", "root")
      .option("dbtable", "tnb_mei")
      .mode(SaveMode.Overwrite)
      .save()
    qianqiDF.write
      .format("jdbc")
      .option("url", "jdbc:mysql://localhost/health_monitoring")
      .option("user", "root")
      .option("password", "root")
      .option("dbtable", "tnb_mei")
      .mode(SaveMode.Append)
      .save()
    zhongduDF.write
      .format("jdbc")
      .option("url", "jdbc:mysql://localhost/health_monitoring")
      .option("user", "root")
      .option("password", "root")
      .option("dbtable", "tnb_mei")
      .mode(SaveMode.Append)
      .save()
    ////    var resultDF: DataFrame=result.toDF()
    //    // 输出分析的结果
    //    //    result.foreach(println)
    //
    //    //6. 将分析结果写入文件
    //    //     6.1 分析结果转换为标准的json格式的字符串
    //    //     6.2 PrintWriter 写文件
    //    /*val jsonResult = "data" -> result.collect().toList.map {
    //
    //             case (propertyname, count) =>
    //               ("agestage", propertyname) ~
    //                 ("count", count)
    //
    //      }*/
    //
    //
    //    val jsonResult = "正常" -> zhengchangresult.collect().toList.map {
    //    data =>
    //      data match {
    //        case (name, value) =>
    //          ("name", name) ~
    //            ("value", value)
    //
    //      }
    //  }
    //    val jsonResult1 =
    //      "糖尿病前期患者" -> qianqiresult.collect().toList.map {
    //    data =>
    //      data match {
    //        case (name, value) =>
    //          ("name", name) ~
    //            ("value", value)
    //
    //      }
    //  }
    //    val jsonResult2 =
    //      "糖尿病重度患者" -> zhongduresult.collect().toList.map {
    //    data =>
    //      data match {
    //        case (name, value) =>
    //          ("name", name) ~
    //            ("value", value)
    //
    //      }
    //  }

    ////
    //    // 写文件
    //    // Array((),())
    //    val outputStream = new PrintWriter(outputFile)
    //    outputStream.write(compact(render(jsonResult)))
    //    outputStream.write(compact(render(jsonResult1)))
    //    outputStream.write(compact(render(jsonResult2)))
    //    outputStream.flush()
    //    outputStream.close()
    //  }
  }
}
